package com.ry.flink.jobx;

import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.types.Row;

/**
 *  3. 输入数据可能会发生update的情况
 * */
public class Job3 {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        StreamTableEnvironment tablEnv = StreamTableEnvironment.create(env);

        //"貂蝉", "孙尚香", "大乔和小乔"
        DataStreamSource<Row> dataStreamSource = env.fromElements(
                Row.of("貂蝉", 85L),
                Row.of("孙尚香", 84L),
                Row.of("貂蝉", 83L));
        Table inputTable = tablEnv.fromDataStream(dataStreamSource).as("name", "score");

        tablEnv.createTemporaryView("dream",inputTable);

        Table table = tablEnv.sqlQuery("select name,sum(score) from dream group by name");

        /**
         * 针对 Retract 流和Upsert 流
         * */
        DataStream<Row> rowDataStream = tablEnv.toChangelogStream(table);
        /**
         *   +I[孙尚香, 84]
         *  +I[貂蝉, 85]
         *  -U[貂蝉, 85]
         *  +U[貂蝉, 168]
         * */
        rowDataStream.print();

        env.execute("Job3");
    }
}
